package org.example.confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.example.constant.Constants;

import java.io.IOException;
import java.util.*;

/**
 * 为了防止出现消息丢失
 * 引入了消息确认应答的这种方式
 */
public class PublisherConfirms {
    public final static Integer MESSAGE_COUNT1=200;
    public static Connection getConnection() throws Exception{
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost(Constants.HOST);
        connectionFactory.setPort(Constants.PORT); //需要提前开放端口号
        connectionFactory.setUsername(Constants.USER_NAME);//账号
        connectionFactory.setPassword(Constants.PASSWORD);  //密码
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机
        return  connectionFactory.newConnection();
    }
    public static void main(String[] args) throws Exception {
        //Strategy #1: Publishing Messages Individually
        //单独确认
        //publishingMessagesIndividually();
        //Strategy #2: Publishing Messages in Batches
        //批量确认
        //publishingMessagesInBatches();

        //Strategy #3: Handling Publisher Confirms Asynchronously
        //异步确认
        handlingPublisherConfirmsAsynchronously();
    }

    /**
     * 异步确认
     */
    private static void handlingPublisherConfirmsAsynchronously() throws Exception {
        try(Connection connection=getConnection()){
            Channel channel = connection.createChannel();
            //1.设置信道为confirm模式
            channel.confirmSelect();
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3,true,false,false,null);
            long start=System.currentTimeMillis();
            SortedSet<Long> sqlSet= Collections.synchronizedSortedSet(new TreeSet<>());
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple) {
                        sqlSet.headSet(deliveryTag+1).clear();
                    } else {
                        sqlSet.remove(deliveryTag);
                    }
                }
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    if(multiple) {
                        //从头开始拿到目标的值
                        //+1是为了小于等于deliveryTag
                        sqlSet.headSet(deliveryTag+1).clear();
                    } else {
                        sqlSet.remove(deliveryTag);
                    }
                }
            });
            for(int i =0;i<MESSAGE_COUNT1;i++) {
                String msg = "hello publisher confirms"+i;
                //发送每个消息的时候，独有的id
                long id=channel.getNextPublishSeqNo();
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3,null,msg.getBytes());
                sqlSet.add(id);
            }
            while(!sqlSet.isEmpty()) {
                Thread.sleep(10);
            }
            long end=System.currentTimeMillis();
            System.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT1, end-start);
        }
    }

    /**
     * 批量确认
     */
    private static void publishingMessagesInBatches() throws Exception {
        try(Connection connection = getConnection()) {
            //1. 开启信道
            Channel channel = connection.createChannel();
            //2. 设置信道为confirm模式
            channel.confirmSelect();
            //3. 声明队列
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);
            //4. 发送消息, 并进行确认
            long start = System.currentTimeMillis();
            int batchSize = 100;
            int outstandingMessageCount = 0;
            for (int i = 0; i < MESSAGE_COUNT1; i++) {
                String msg = "hello publisher confirms"+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());
                outstandingMessageCount++;
                if (outstandingMessageCount==batchSize){
                    channel.waitForConfirmsOrDie(5000);
                    outstandingMessageCount = 0;
                }
            }
            if (outstandingMessageCount>0){
                channel.waitForConfirmsOrDie(5000);
            }
            long end = System.currentTimeMillis();
            System.out.printf("批量确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT1, end-start);
        }
    }

    /**
     * 单独确认
     */
    private static void publishingMessagesIndividually() throws Exception {
        try(Connection connection=getConnection()){
            Channel channel = connection.createChannel();
            //1.设置信道为confirm模式
            channel.confirmSelect();
            channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1,true,false,false,null);
            long start=System.currentTimeMillis();
            for( int i=0;i<MESSAGE_COUNT1;i++) {
                String msg = "hello publisher confirms"+i;
                channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1,null,msg.getBytes());
                channel.waitForConfirmsOrDie(5000);
            }
            long end=System.currentTimeMillis();
            System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT1, end-start);
        }
    }
}
